/**
 * 
 */
package com.db.stockexchange.server;

import java.util.concurrent.BlockingQueue;
import java.util.logging.Logger;

import com.db.stockexchange.common.OrderType;
import com.db.stockexchange.common.messages.FIXMessage;
import com.db.stockexchange.common.messages.client.*;
import com.db.stockexchange.common.messages.server.*;
import com.db.stockexchange.server.orderbook.Deal;
import com.db.stockexchange.server.orderbook.Order;

/**
 * @author Rayman
 * Transform messages from client to method of business logic calls and vice versa
 */
public class MessageHandler implements Runnable {
	static final int SLEEP_TIMEOUT = 500;
	
	StockExchangeEngine stockExchange;	
	BlockingQueue<FIXMessage> fromEngineQueue;
	BlockingQueue<FIXMessage> toEngineQueue;
	Logger logger;
	
	volatile boolean working = true;
	
	public MessageHandler(BlockingQueue<FIXMessage> fromEngine, BlockingQueue<FIXMessage> toEngine) {
		this.fromEngineQueue = fromEngine;
		this.toEngineQueue = toEngine;
		logger = Logger.getLogger(this.getClass().getName());
		
		new Thread(this, "MessageHandler").start();
	}
	
	public void setStockExchange(StockExchangeEngine engine) {
		this.stockExchange = engine;
	}
	
	private void sendMessage(FIXMessage message) {
		try {
			fromEngineQueue.put(message);
		} catch (InterruptedException e) { }
	}

	private void processMessage(FIXMessage message) {
		logger.fine(String.format("Process message with type <%s>", message.getType()));
		switch (message.getType()) {
		case NEW_ORDER:
			NewOrderMessage m = (NewOrderMessage)message;
			Order newOrder;
			if (m.getOrderType() == OrderType.MARKET)
				newOrder = new Order(m.getOrderID(), m.getSenderCompID(), m.getSymbol(), 
						m.getOrderQuantity(), m.getSide());
			else
				newOrder = new Order(m.getOrderID(), m.getSenderCompID(), m.getSymbol(), 
						m.getOrderQuantity(), m.getPrice(), m.getSide());
			
			if (!stockExchange.isTickerNameValid(newOrder.getTickerName())) {
				logger.warning(String.format("Could not find ticker by name <%s>", newOrder.getTickerName()));
				sendReject(newOrder, "Could not find ticker by name <" + newOrder.getTickerName() + ">");
			} else			
				stockExchange.addNewOrder(newOrder);
			
			break;
			default:
				logger.warning(String.format("Message with type <%s> is unsupported", message.getType()));
		}
	}
	
	public void sendExecReport(Deal deal) {
		Order ask = deal.getAskOrder();
		Order bid = deal.getBidOrder();
		FIXMessage m1 = new FilledMessage(ask.getClientId(), ask.getId(), ask.getTickerName(), ask.getSide(), ask.getStatus(), 
								ask.getOrderQuantity(), ask.getCumQuantity(), ask.getLeavesQuantity(), deal.getPrice().getValue(),
								deal.getPrice().getValue(), deal.getQuantity(), bid.getClientId()); 
		
		FIXMessage m2 = new FilledMessage(bid.getClientId(), bid.getId(), bid.getTickerName(), bid.getSide(), bid.getStatus(), 
								bid.getOrderQuantity(), bid.getCumQuantity(), bid.getLeavesQuantity(), deal.getPrice().getValue(),
								deal.getPrice().getValue(), deal.getQuantity(), ask.getClientId()); 
		
		sendMessage(m1);
		sendMessage(m2);
	}
	
	public void sendReject(Order order, String reason) {
		FIXMessage m = new ExecutionRejectMessage(order.getClientId(), order.getId(), order.getTickerName(), order.getSide(), reason);
		sendMessage(m);
	}

	public void sendCancel(Order order) {
		throw new UnsupportedOperationException();
	}
	
	@Override
	public void run() {
		// processing messages to engine from clients
		while (working) {
			FIXMessage message = null;
			try {
				message = toEngineQueue.take();
			} catch (InterruptedException e) { }
			if (message != null && message.getClass() != PoisonMessage.class)
				processMessage(message);
		}
		
		dispose();
	}
	
	public void close() {
		working = false;
		toEngineQueue.add(new PoisonMessage());
	}
	
	private void dispose() {
		toEngineQueue.clear();
	}
}
